package fun.tan90.device.source.mysql.plugin.util;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import com.github.shyiko.mysql.binlog.event.deserialization.json.JsonBinary;
import fun.tan90.device.source.mysql.plugin.model.ColumnMetadata;
import lombok.Getter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.noear.snack.ONode;

import java.io.Serializable;
import java.sql.Date;
import java.sql.Time;
import java.sql.Timestamp;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;

@Slf4j
@Getter
public class BinlogUtil {

    public static ONode toNode(List<ColumnMetadata> columns, Serializable[] data, long timeOffset) {
        return ONode.load(toMap(columns, data, timeOffset));
    }

    @SneakyThrows
    private static Map<String, Object> toMap(List<ColumnMetadata> columns, Serializable[] data, long timeOffset) {
        Map<String, Object> resMap = new HashMap<>(columns.size());
        for (int i = 0; i < Integer.min(columns.size(), data.length); i++) {
            Serializable serializable = data[i];
            ColumnMetadata columnMetadata = columns.get(i);
            String columnName = columnMetadata.getColumnName();
            //log.info("columnName: {}, columnDataType: {}, serializable: {}", columnName, columnMetadata.getDataType(), Objects.isNull(serializable) ? null : serializable.getClass());
            if (columnMetadata.getDataType().equalsIgnoreCase("json") && serializable instanceof byte[] bytes) {
                resMap.put(columnName, JsonBinary.parseAsString(bytes));
            } else if (columnMetadata.getDataType().equalsIgnoreCase("time") && serializable instanceof Time time) {
                resMap.put(columnName, new Time(time.getTime() + timeOffset * 1000L).toString());
            } else if (columnMetadata.getDataType().equalsIgnoreCase("timestamp") && serializable instanceof Timestamp timestamp) {
                resMap.put(columnName, DateUtil.format(timestamp, DatePattern.NORM_DATETIME_MS_PATTERN));
            } else if (columnMetadata.getDataType().equalsIgnoreCase("date") && serializable instanceof Date date) {
                resMap.put(columnName, DateUtil.format(date, DatePattern.NORM_DATE_PATTERN));
            } else if (columnMetadata.getDataType().equalsIgnoreCase("datetime") && serializable instanceof java.util.Date date) {
                resMap.put(columnName, DateUtil.format(new Date(date.getTime() + timeOffset * 1000L), DatePattern.NORM_DATETIME_MS_PATTERN));
            } else if (columnMetadata.getDataType().endsWith("text") && serializable instanceof byte[] bytes) {
                resMap.put(columnName, new String(bytes));
            } else {
                resMap.put(columnName, serializable);
            }
        }
        return resMap;
    }
}
